package com.funny.rabbitmq.deadqueue;

import com.funny.rabbitmq.utils.RabbitMQUtils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;

import java.util.HashMap;
import java.util.Map;

@Slf4j
public class Consumer {

    // 普通交换机和队列
    private final static String NORMAL_EXCHANGE_NAME = "normal.exchange";
    private final static String NORMAL_QUEUE_NAME = "normal.queue";

    // 死信交换机和队列
    private final static String DEAD_EXCHANGE_NAME = "dead.exchange";
    private final static String DEAD_QUEUE_NAME = "dead.queue";

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMQUtils.getChannel();

        // 声明交换机
        channel.exchangeDeclare(NORMAL_EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
        channel.exchangeDeclare(DEAD_EXCHANGE_NAME, BuiltinExchangeType.DIRECT);

        // 声明队列
        Map<String, Object> arguments = new HashMap<>();
        // 过期时间，毫秒,在发送方设置时间较灵活
//        arguments.put("x-message-ttl", 10000);
        // 正常队列设置死信交换机
        arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE_NAME);
        // 死信routingKey
        arguments.put("x-dead-letter-routing-key", "lisi");
        // 正常队列的长度
//        arguments.put("x-max-length", 6);
        channel.queueDeclare(NORMAL_QUEUE_NAME, false, false, false, arguments);
        channel.queueBind(NORMAL_QUEUE_NAME, NORMAL_EXCHANGE_NAME, "zhangsan");

        channel.queueDeclare(DEAD_QUEUE_NAME, false, false, false, null);
        channel.queueBind(DEAD_QUEUE_NAME, DEAD_EXCHANGE_NAME, "lisi");

        channel.basicConsume(NORMAL_QUEUE_NAME, false, (consumerTag, message) -> {
            String msg = new String(message.getBody());
            String[] strs = msg.split("_");
            int n = Integer.parseInt(strs[1]);

            log.info("consumerTag={}, message={}", consumerTag, msg);
            long deliveryTag = message.getEnvelope().getDeliveryTag();
            String routingKey = message.getEnvelope().getRoutingKey();
            log.info("deliveryTag={}, routingKey={}", deliveryTag, routingKey);
            log.info("n={}", n);
            if (n == 5) {
                channel.basicReject(deliveryTag, false);
            }
        }, consumerTag -> {
            log.info("consumerTag={}", consumerTag);
        });

    }

    private static boolean handleMsg(String msg) {
        String[] strs = msg.split("_");
        int n = Integer.parseInt(strs[1]);
        return n > 2;
    }
}
